package streamproc;

import java.util.HashMap;
import java.util.Vector;
import streamdef.StreamCondition;
import streamdef.StreamDefinitionManager;
import streamdef.EventSet;
import event.Event;

public class SequenceDetectorComponent extends ChainComponent {
	private class StackObject {
		public StackObject() {
			event = null;
			prevObjectInStack = null;
			lastObjectInPrevStack = null;
			nextObjectInStack = null;
		}

		Event event;
		StackObject prevObjectInStack;
		StackObject nextObjectInStack;
		StackObject lastObjectInPrevStack;
	}

	private class SequenceState {
		public SequenceState() {
			streamIDOfSequence = -1;
			internalStreamIDOfState = -1;
			rangeInMilliseconds = -1;
			prevState = null;
			top = null;
		}

		long streamIDOfSequence;
		long internalStreamIDOfState;
		long rangeInMilliseconds;
		SequenceState prevState;

		StackObject top;
	}

	private static SequenceDetectorComponent sInstance = null;

	public static SequenceDetectorComponent instance() {
		if (sInstance == null)
			sInstance = new SequenceDetectorComponent();

		return sInstance;
	}

	private HashMap<Long, Vector<SequenceState>> streamIDToEndStates = null;
	private HashMap<Long, Vector<SequenceState>> streamIDToInBetweenStates = null;

	private SequenceDetectorComponent() {
		streamIDToEndStates = new HashMap<Long, Vector<SequenceState>>();
		streamIDToInBetweenStates = new HashMap<Long, Vector<SequenceState>>();
	}

	public void reset() {
		if (streamIDToEndStates != null)
			streamIDToEndStates.clear();
		if (streamIDToInBetweenStates != null)
			streamIDToInBetweenStates.clear();
	}

	protected void finalize() {
		reset();
	}

	public void init() {
		reset();
		Vector<Long> streamIDs = StreamDefinitionManager.instance()
				.getStreamIDs();

		for (int i = 0; i < streamIDs.size(); ++i) {
			long sID = streamIDs.get(i);
			StreamCondition mq = StreamDefinitionManager.instance()
					.getParsedQuery(sID);
			Vector<Vector<Long>> possSequences = mq.getPossibleEventSequences();
			for (int j = 0; j < possSequences.size(); ++j) {
				Vector<Long> seq = possSequences.get(j);
				SequenceState prevSS = null;

				for (int k = 0; k < seq.size(); ++k) {
					long intStateID = seq.get(k);
					long extStateID = mq.getExternalStreamID(intStateID);

					SequenceState ss = new SequenceState();
					ss.streamIDOfSequence = sID;
					ss.internalStreamIDOfState = intStateID;
					ss.rangeInMilliseconds = (long) (mq
							.getRangeForInternalStreamID(intStateID) * 1000);
					ss.prevState = prevSS;

					if (k != seq.size() - 1) {
						if (!streamIDToInBetweenStates.containsKey(extStateID))
							streamIDToInBetweenStates.put(extStateID,
									new Vector<SequenceState>());
						streamIDToInBetweenStates.get(extStateID).add(ss);
					} else {
						if (!streamIDToEndStates.containsKey(extStateID))
							streamIDToEndStates.put(extStateID,
									new Vector<SequenceState>());
						streamIDToEndStates.get(extStateID).add(ss);
					}

					prevSS = ss;
				}
			}
		}
	}

	public synchronized void processEvent(Event e) {
		long extID = e.getStreamID();

		//
		// find sequences for when <e> is considered as a new endState
		//
		Vector<EventSet> foundSequences = new Vector<EventSet>();
		Vector<SequenceState> endStates = this.streamIDToEndStates.get(extID);

		if (endStates != null) {
			for (int i = 0; i < endStates.size(); ++i) {
				SequenceState seqStat = endStates.get(i);

				StackObject so = new StackObject();
				if (seqStat.top != null)
					seqStat.top.nextObjectInStack = so;
				so.prevObjectInStack = seqStat.top;
				so.event = e;
				if (seqStat.prevState != null) {
					so.lastObjectInPrevStack = seqStat.prevState.top;
				}

				seqStat.top = so;

				foundSequences.addAll(constructSequencesFromEndState(seqStat, e
						.getTimestamp().getTime()));
			}
		}

		//
		// update inbetweenstates
		//
		Vector<SequenceState> inBetweenStates = streamIDToInBetweenStates
				.get(extID);

		if (inBetweenStates != null) {
			// start at the back -> update from endstate to beginstate
			for (int i = inBetweenStates.size() - 1; i >= 0; --i) {
				SequenceState stat = inBetweenStates.get(i);

				StackObject so = new StackObject();
				if (stat.top != null)
					stat.top.nextObjectInStack = so;
				so.prevObjectInStack = stat.top;
				so.event = e;
				if (stat.prevState != null) {
					so.lastObjectInPrevStack = stat.prevState.top;
				}

				stat.top = so;
			}
		}

		//
		// push foundSequences
		//
		if (endStates != null)
			next.processQueryResults(foundSequences);
	}

	private Vector<EventSet> constructSequencesFromEndState(
			SequenceState seqStat, long timestampEndEvent) {
		long streamIDSequence = seqStat.streamIDOfSequence;
		Vector<EventSet> results = new Vector<EventSet>();

		EventSet startQr = new EventSet(streamIDSequence);
		startQr.addEvent(seqStat.internalStreamIDOfState, seqStat.top.event);

		SequenceState startSeq = seqStat.prevState; // state before seqStat
		// most recent object in the stack of startSeq before seqStat.top
		StackObject startStackObject = seqStat.top.lastObjectInPrevStack;

		constructSequencesToBeginState(results, startQr, startSeq,
				startStackObject, timestampEndEvent);

		// top of last stack must be processed once
		// -> no need for an actual stack
		seqStat.top = null;

		return results;
	}

	private void constructSequencesToBeginState(Vector<EventSet> results,
			EventSet startQr, SequenceState startSeq,
			StackObject startStackObject, long timestampEndEvent) {
		if (startSeq == null) {
			// now we are before the beginning of a sequence
			// -> end of recursion
			results.add(startQr);
		} else {
			StackObject so = startStackObject;
			SequenceState prevSS = startSeq.prevState;

			while (so != null
					&& (timestampEndEvent - so.event.getTimestamp().getTime()) <= startSeq.rangeInMilliseconds) {
				EventSet qr = startQr.clone();
				qr.addEvent(startSeq.internalStreamIDOfState, so.event);
				// recursive call
				constructSequencesToBeginState(results, qr, prevSS,
						so.lastObjectInPrevStack, timestampEndEvent);
				so = so.prevObjectInStack;
			}

			// StackObject so is outdated for the current endEvent
			if (so != null)
				removePreviousEventsFromStack(so);
		}
	}

	// set all fields to null of the outdated StackObjects
	// -> help the garbage collector in its task
	private void removePreviousEventsFromStack(StackObject firstOutdatedEvent) {
		// first stack in sequence reached
		if (firstOutdatedEvent.lastObjectInPrevStack == null) {
			// empty till stack bottom
			StackObject so = firstOutdatedEvent;
			while (so != null) {
				if (so.nextObjectInStack != null) {
					so.nextObjectInStack.prevObjectInStack = null;
					so.nextObjectInStack = null;
				}

				so.lastObjectInPrevStack = null;
				so = so.prevObjectInStack;
			}
		} else {
			// recursive to the first stack in the sequence
			removePreviousEventsFromStack(firstOutdatedEvent.lastObjectInPrevStack);

			// empty current stack till bottom
			StackObject so = firstOutdatedEvent;
			while (so != null) {
				if (so.nextObjectInStack != null) {
					so.nextObjectInStack.prevObjectInStack = null;
					so.nextObjectInStack = null;
				}

				so.lastObjectInPrevStack = null;
				so = so.prevObjectInStack;
			}
		}
	}
}
